package org.future.task.core.executor;

import org.future.task.core.handler.Commit;
import org.future.task.core.handler.Connection;
import org.future.task.core.util.ThreadPoolUtils;

import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

public class FixedExecutor implements Executor {

    private final ThreadPoolExecutor threadPoolExecutor;

    private Connection connection;

    private int writeBatch;

    public FixedExecutor(ThreadPoolExecutor threadPoolExecutor, Connection connection) {
        if (threadPoolExecutor == null) {
            threadPoolExecutor = new ThreadPoolUtils().getDefaultThreadPool("commit", 1);
        }
        this.threadPoolExecutor = threadPoolExecutor;
        this.connection = connection;
    }

    public FixedExecutor(String poolType, ThreadPoolExecutor threadPoolExecutor, Connection connection) {
        if (threadPoolExecutor == null) {
            threadPoolExecutor = new ThreadPoolUtils().getDefaultThreadPool(poolType, 1);
        }
        this.threadPoolExecutor = threadPoolExecutor;
        this.connection = connection;
    }

    public int getWriteBatch() {
        return writeBatch;
    }

    public void setWriteBatch(int writeBatch) {
        this.writeBatch = writeBatch;
    }

    @Override
    public void submit(String queueName, Commit commit, int threadSize) {
        if (threadSize <= 0) {
            threadSize = 1;
        }
        synchronized (threadPoolExecutor) {
            threadPoolExecutor.setCorePoolSize(threadPoolExecutor.getPoolSize() + threadSize);
        }
        for (int i = 0; i < threadSize; i++) {
            threadPoolExecutor.execute(new CommitRunnable(queueName, commit, connection));
        }
    }

    class CommitRunnable implements Runnable {

        private final String queueName;

        private final Commit commit;

        private final Connection connection;

        public CommitRunnable(String queueName, Commit commit, Connection connection) {
            this.queueName = queueName;
            this.commit = commit;
            this.connection = connection;
        }

        @Override
        public void run() {

            while (true) {
                try {
                    List batch = commit.getBatch(queueName, writeBatch <= 0 ? 100 : writeBatch);
                    if (batch == null || batch.isEmpty()) {
                        continue;
                    }
                    connection.sendDataList(queueName, batch);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

        }
    }

}
